package org.homonoia.worker.core;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.AbstractMap;
import java.util.AbstractMap.SimpleEntry;
import java.util.Properties;

import org.apache.log4j.Logger;
import org.homonoia.model.core.Column;
import org.homonoia.model.core.Row;
import org.homonoia.model.query.QueryID;
import org.homonoia.model.query.QueryResults;
import org.homonoia.model.task.Task;
import org.homonoia.model.task.TaskResults;
import org.homonoia.model.task.TaskType;
import org.homonoia.worker.sql.SearchMySQL;
import org.homonoia.worker.sql.SearchSQL;

public class TaskRunner implements Runnable {

    private static final Logger log = Logger.getLogger(TaskRunner.class);

    private WorkerNodeService workerNodeService;
    private Task task;
    private QueryID id;

    public TaskRunner(WorkerNodeService workerNodeService, Task task, QueryID id) {
	this.workerNodeService = workerNodeService;
	this.task = task;
	this.id = id;
    }

    public void run() {
	runTask();
	workerNodeService.setTaskCompleted();
    }

    public void runTask() {
	log.info("Running: " + task.getID());

	if (task.getTaskType() == TaskType.SEARCH) {
	    runSearchTask();
	} else if (task.getTaskType() == TaskType.JOIN) {
	    runJoinTask();
	}

	log.info("Completed: " + task.getID());
    }

    private void runJoinTask() {
	QueryResults results = new QueryResults();

	for (TaskResults res : task.getTaskResults()) {
	    results.appendTaskResults(res);
	}

	Socket sock = null;
	PrintWriter out = null;
	ObjectOutputStream oos = null;

	try {
	    sock = new Socket(workerNodeService.getConfiguration().getProperty("Controller"),
		    Integer.parseInt(workerNodeService.getConfiguration().getProperty(
			    "ControllerPort")));

	    out = new PrintWriter(sock.getOutputStream(), true);
	    out.println("Request: application/join");

	    oos = new ObjectOutputStream(sock.getOutputStream());
	    AbstractMap.SimpleEntry<QueryID, QueryResults> pair = new SimpleEntry<QueryID, QueryResults>(
		    id, results);
	    oos.writeObject(pair);

	} catch (NumberFormatException e) {
	    log.error(e.getMessage(), e);
	} catch (UnknownHostException e) {
	    log.error(e.getMessage(), e);
	} catch (IOException e) {
	    log.error(e.getMessage(), e);
	}
    }

    private void runSearchTask() {
	int colCount = -1;

	Connection conn = null;
	Properties connectionProps = new Properties();
	connectionProps.put("user", workerNodeService.getConfiguration().getProperty("DBUser"));
	connectionProps.put("password", workerNodeService.getConfiguration().getProperty("DBPass"));

	String url = workerNodeService.getConfiguration().getProperty("DBUrl");
	try {
	    conn = DriverManager.getConnection("jdbc:" + url, connectionProps);
	    Statement stmt = conn.createStatement();
	    SearchSQL sqlSearch = null;

	    if (url.contains("mysql")) {
		sqlSearch = new SearchMySQL();
	    }

	    sqlSearch.checkForRowIDColumn(stmt, task);
	    colCount = sqlSearch.getColumnCount(stmt, task);
	    String sql = sqlSearch.generateStatement(task);

	    ResultSet rs = stmt.executeQuery(sql);
	    TaskResults results = parseStatementResults(rs, colCount);

	    if (stmt != null) {
		stmt.close();
	    }

	    if (conn != null) {
		conn.close();
	    }

	    sendTaskResults(results);

	} catch (SQLException e) {
	    log.error(e.getMessage(), e);
	}
    }

    private TaskResults parseStatementResults(ResultSet rs, int colCount) throws SQLException {
	TaskResults results = new TaskResults(task.getID());
	String value = null;
	while (rs.next()) {
	    Row r = new Row();
	    for (int i = 1; i <= colCount; ++i) {
		value = rs.getString(i);
		r.insertColumn(new Column(value));
		results.insertRow(r);
	    }
	}
	return results;
    }

    private void sendTaskResults(TaskResults results) {
	Socket sock = null;
	PrintWriter out = null;
	ObjectOutputStream oos = null;

	try {
	    sock = new Socket(workerNodeService.getConfiguration().getProperty("Controller"),
		    Integer.parseInt(workerNodeService.getConfiguration().getProperty(
			    "ControllerPort")));

	    out = new PrintWriter(sock.getOutputStream(), true);
	    out.println("Request: application/search");

	    oos = new ObjectOutputStream(sock.getOutputStream());
	    oos.flush();
	    AbstractMap.SimpleEntry<QueryID, TaskResults> pair = new SimpleEntry<QueryID, TaskResults>(
		    id, results);
	    oos.writeObject(pair);
	} catch (NumberFormatException e) {
	    log.error(e.getMessage(), e);
	} catch (UnknownHostException e) {
	    log.error(e.getMessage(), e);
	} catch (IOException e) {
	    log.error(e.getMessage(), e);
	}
    }
}
